package cn.edu.xmu.datamining.tangzk.mralgos.sssp;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import cn.edu.xmu.datamining.tangzk.util.MyUtil;

public class SingleSourceShortestPathReducer extends
		Reducer<IntWritable, GraphNodeWritable, IntWritable, Text> {
	protected void reduce(
			IntWritable key,
			java.lang.Iterable<GraphNodeWritable> values,
			org.apache.hadoop.mapreduce.Reducer<IntWritable, GraphNodeWritable, IntWritable, Text>.Context context)
			throws java.io.IOException, InterruptedException {

		int mindist = Integer.MAX_VALUE;
		GraphNodeWritable struct = null;
		for (GraphNodeWritable node : values) {
			if (node.isNode()) {
				struct = node.clone();
			} else {
				mindist = Math.min(mindist, node.getDistance());
			}
		}
		if (struct == null) {
			// non-struct information output
			struct = new GraphNodeWritable(Integer.MAX_VALUE, "");
		}
		if (mindist < struct.getDistance()) {
			struct.setDistance(mindist);
			context.getCounter(
					SingleSourceShortestPathDriver.TERM_COUNTER_GROUP,
					SingleSourceShortestPathDriver.TERM_COUNTER_NAME)
					.increment(1);
		}
		StringBuilder sb = new StringBuilder();
		sb.append(MyUtil.formatInt(struct.getDistance()));

		if (struct.getEdgeList() != null) {
			sb.append("\t");
			boolean first = true;
			for (GraphEdge edge : struct.getEdgeList()) {
				if (first) {
					first = false;
				} else {
					sb.append(",");
				}
				sb.append(edge.toString());
			}
		}
		context.write(key, new Text(sb.toString()));
	};
}
